/*
Code : Dzlua
Email: 505544956@qq.com
Time : 2017/06/05
*/
#ifndef DZLUA_CS_LIBEVENT_SERVER_H_
#define DZLUA_CS_LIBEVENT_SERVER_H_

#pragma once

#include <event.h>
#include <event2/listener.h>
#include <event2/thread.h>

#include <thread>
#include <vector>
#include <mutex>

#if defined(WIN32) || defined(_WIN32)
#   define DZLUA_OS_WINDOWS 1
#endif

namespace dzlua {

namespace cs {

#define DZLUA_CS_EXITCODE -1

class LibeventServer;
class LibeventConn;
struct LibeventThread;

struct LibeventThread {
    // thread.
    std::shared_ptr<std::thread> td;

    /*
        Many callback functions are used in the event handling of libevent,
            and class implicit this pointers cannot be used,
            So pass the pointer of TcpBaseServer in this way.
    */
    LibeventServer *server;

    // event handler of libevent.
    struct event_base *base;

    // listener managed event.
    struct event notify;

    // Receiving end of pipe.
    evutil_socket_t recv;

     // Sending end of pipe.
    evutil_socket_t send; 
};

/* 
    1. This class is a list of nodes, each connected node in the information storage,
        and provides read and write data interface.
    2. This class must be created by TcpBaseServer, and managed by the class ConnQueue.
*/
class LibeventConn {
    friend class LibeventThread;
public:
    LibeventThread *GetThread() {
        return m_Thread;
    }

    int GetFd() {
        return m_fd;
    }

    // get the length of the readable data.
    int GetReadBufferLen() {
        return evbuffer_get_length(m_ReadBuf);
    }

    /*
        Remove len bytes from the read buffer and stored in buffer,
            if not enough, then read all the data.
        Return the number bytes of data read.
    */
    int GetReadBuffer(char *buffer, int len) {
        return evbuffer_remove(m_ReadBuf, buffer, len);
    }

    /*
        Copy len bytes from the read buffer and stored in buffer,
            if not enough, then all the data is copied.
        Return the number bytes of data copied.
        When the operation is performed, the data is left in the buffer,
            and the data in buffer is only a copy of the original data.
    */
    int CopyReadBuffer(char *buffer, int len) {
        return evbuffer_copyout(m_ReadBuf, buffer, len);
    }

    // Get the length of writable data.
    int GetWriteBufferLen() {
        return evbuffer_get_length(m_WriteBuf);
    }

    // Adding data to the write buffer, ready to send.
    int AddToWriteBuffer(char *buffer, int len) {
        return evbuffer_add(m_WriteBuf, buffer, len);
    }

    // Move the data in the read buffer to the write buffer.
    void MoveBufferData() {
        evbuffer_add_buffer(m_WriteBuf, m_ReadBuf);
    }
private:
    LibeventConn(const int fd, LibeventThread *td)
            : m_fd(fd)
            , m_ReadBuf(nullptr)
            , m_WriteBuf(nullptr)
            , m_Thread(td) {
    }
private:
    const int m_fd;             // id of socket
    evbuffer *m_ReadBuf;        // read data buffer
    evbuffer *m_WriteBuf;       // write data buffer
    LibeventThread *m_Thread;   // thread info
};

class LibeventServer {
public:
    typedef void (*funTimerEvent)(int id, short events, LibeventServer *server);
    typedef void (*funSignalEvent)(int sig, short events, LibeventServer *server);
public:
    LibeventServer()
            , m_port(DZLUA_CS_EXITCODE)
            , m_MainBase(nullptr)
            , m_Threads(nullptr) {
        m_SignalEvents.clear();
    }

    ~LibeventServer() {}
public:

    /*
        Set thread counts and port
            and init the LibeventServer.
    */
    bool Init(int count, int port) {
        m_port = port;
#if defined(DZLUA_OS_WINDOWS)
        {
            WORD wVersion;
            WSADATA wsaData;
            wVersion = MAKEWORD(2, 2);
            int err = WSAStartup(wVersion, &wsaData);
            if (0 != err) {
                perror("error init.\n");
                return false;
            }

            if (wsaData.wVersion != wVersion) {
                perror("error init version.\n");
                WSACleanup();
                return false;
            }
        }
        if (evthread_use_windows_threads() < 0) {
            perror("error init evthread_use_windows_threads.\n");
            goto ERROR;
        }
#else
        if (evthread_use_pthreads() < 0) {
            perror("error init evthread_use_pthreads.\n");
            goto ERROR;
        }
#endif // end DZLUA_OS_WINDOWS

        m_MainBase = std::move(
            std::unique_ptr<LibeventThread>(
                new LibeventThread ));
        m_MainBase->td = nullptr;
        m_MainBase->base = event_base_new();
        if (!m_MainBase || !m_MainBase->base) {
            perror("error init new LibeventThread or error event_base_new.\n");
            goto Error;
        }

        // for all threads
        for (int i = 0; i < count; ++i) {
            auto ltd = std::unique_ptr<LibeventThread>(
                new LibeventThread );
            if (!SetupThread(ltd.get()));
                goto Error;
            m_Threads.push_back(std::move(ltd));
        }
        return true;
    Error:
        for (auto it = m_Threads.begin(); it != m_Threads.end(); ++it) {
            event_base_free((*it)->base);
        }
#if defined(DZLUA_OS_WINDOWS)
        WSACleanup();
#endif
        return false;
    }

    // Start event loop.
    bool StartRun() {
        if (DZLUA_CS_EXITCODE == m_Port) {
            perror("Error : StartRun:: error port.");
            return false;
        }
        
        // listen to the port number.
        {
            sockaddr_in sin;
            memset(&sin, 0, sizeof(sin));
            sin.sin_family = AF_INET;
            sin.sin_port = htons(m_port);
            listener = evconnlistener_new_bind(
                            m_MainBase->base,
                            ListenerEventCb, (void*)this,
                            LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE, -1,
                            (sockaddr*)&sin, sizeof(sockaddr_in));
            if( nullptr == listener ) {
                fprintf(stderr, "listen error: %s\n", strerror(errno));
                return false;
            }
        }

        // start each child thread.
        for (auto it = m_Threads.begin(); it != m_Threads.end(); ++it) {
            (*it)->td.reset(new std::thread(
                    [] (LibeventThread* ltd) {
                        event_base_dispatch(ltd->base);
                    }, (*it).get()));
        }

        printf("listening...\n");
        // start The event loop of the primary thread.
        event_base_dispatch(m_MainBase->base);

        // Event loop done, release listener memory.
        evconnlistener_free(listener);
    }

    /*
        End the event loop in tv time.
        if tv is empty, stop immediately.
    */
    void StopRun(timeval *tv);

    /*
        Adding and deleting signal processing events.
        sig is the signal, and ptr is the function to call back.
    */
    bool AddSignalEvent(int sig, funSignalEvent callback);
    bool DeleteSignalEvent(int sig);

    /*
        Add and delete timed events.
        
        ptr is the function to call back,
        tv is the interval,
        once determines whether to execute it only once.
    */
    event *AddTimerEvent(funTimerEvent callback, timeval tv, bool once);
    bool DeleteTImerEvent(event *ev);

protected:
    /*
        It is generally necessary to inherit and deal with the specific business
    */

    // When the new connection is successful, the function is called. 
    virtual void ConnectionEvent(Conn *conn) {}

    // After reading the data, the function is called.
    virtual void ReadEvent(Conn *conn) {}

    /* 
        The function is called after the completion of the call.
        Because of a string problem, it is not called every time the data is sent.
    */
    virtual void WriteEvent(Conn *conn) {}

    /*
        This function is called when the connection is disconnected,
            or the customer is automatically disconnected.
    */
    virtual void CloseEvent(Conn *conn, short events) {}

private:
    // Initializes the data of the child thread.
    bool SetupThread(LibeventThread *ltd) {
        if (!ltd) return false;
        ltd->server = this;

        ltd->base = event_base_new();
        if (ltd->base == nullptr) {
            perror("Error : SetupThread:: event_base_new.\n");
            goto Error;
        }

#if defined(DZLUA_OS_WINDOWS)
        evutil_socket_t fds[2];
        if(evutil_socketpair(AF_INET, SOCK_STREAM, 0, fds) < 0) {
            perror("Error : SetupThread:: evutil_socketpair.\n");
            goto Error;
        }
        evutil_make_socket_nonblocking(fds[0]);
        evutil_make_socket_nonblocking(fds[1]);
#else
        int fds[2];
        if (pipe(fds) < 0) {
            perror("Error : SetupThread:: Can't create notify pipe.\n");
            goto Error;
        }
#endif

        // Let the child thread's state machine listen to the pipe.
        event_set( &ltd->notify, ltd->recv,
                EV_READ | EV_PERSIST, &MultiServer::ThreadProcess, ltd );
        if ( event_base_set(ltd->base, &ltd->notify) < 0) {
            perror("Error : SetupThread:: event_base_set.\n");
            goto Error1;
        }
        if ( event_add(&ltd->notify, nullptr) < 0) {
            perror("Error : SetupThread:: event_add.\n");
            goto Error1;
        }

        return true;
    Error1:
        evutil_closesocket(fds[0]);
        evutil_closesocket(fds[1]);
    Error:
        if (ltd->base)
            event_base_free(ltd->base);
        return false;
    }

    /*
        After the primary thread receives the request.
        It corresponds to the processing function of the child thread.
    */
    static void ThreadProcess(int fd, short which, void *arg) {
        LibeventThread *ltd = (LibeventThread*)arg;
        /*
            Read data from a pipe.
            Descriptor or operation code of socket.
        */
        int pipefd = ltd->notifyReceiveFd;
        evutil_socket_t confd;
        recv(pipefd, (char*)&confd, sizeof(evutil_socket_t), 0);

        /*
            If the opcode is DZLUA_CS_EXITCODE, 
                the event loop will be break.
        */
        if( DZLUA_CS_EXITCODE == confd ) {
            event_base_loopbreak(ltd->base);
            return;
        }

        // make new connection.
        struct bufferevent *bev = nullptr;
        bev = bufferevent_socket_new(ltd->base, confd, BEV_OPT_CLOSE_ON_FREE);
        if (!bev) {
            fprintf(stderr, "Error constructing bufferevent!");
            event_base_loopbreak(ltd->base);
            return;
        }

        // Ready to read and write data from socket
        bufferevent_setcb(bev,
                &MultiServer::ReadEventCb,
                &MultiServer::WriteEventCb,
                &MultiServer::CloseEventCb, conn);
        bufferevent_enable(bev, EV_WRITE);
        bufferevent_enable(bev, EV_READ);

        // Call user-defined connection event handler.
        ltd->server->ConnectionEvent(conn);
    }

    // the callbacks called by libevent
    static void ListenerEventCb(struct evconnlistener *listener
            , evutil_socket_t fd, sockaddr *sa
            , int socklen, void *user_data) {
        LibeventServer *server = (LibeventServer*)user_data;

        /*
            Randomly select a child thread, 
                and pass the socket descriptor through the pipe.
        */
        int num = rand() % server->m_ThreadCount;
        int sendfd = server->m_Threads[num].get()->notifySendFd;
        send(sendfd, (char*)&fd, sizeof(evutil_socket_t), 0);
    }
    
    static void ReadEventCb(struct bufferevent *bev, void *data) {
        LibeventConn *conn = (LibeventConn*)data;
        conn->m_ReadBuf = bufferevent_get_input(bev);
        conn->m_WriteBuf = bufferevent_get_output(bev);

        // Call user-defined read event handler function.
        conn->m_Thread->server->ReadEvent(conn);
    }

    static void WriteEventCb(struct bufferevent *bev, void *data) {
        LibeventConn *conn = (LibeventConn*)data;
        conn->m_ReadBuf = bufferevent_get_input(bev);
        conn->m_WriteBuf = bufferevent_get_output(bev);

        // Call user-defined write event handler functions.
        conn->m_Thread->server->WriteEvent(conn);
    }

    static void CloseEventCb(struct bufferevent *bev, short events, void *data) {
        LibeventConn *conn = (LibeventConn*)data;
        // Call user-defined break event handler function.
        conn->m_Thread->server->CloseEvent(conn, events);
        bufferevent_free(bev);
    }
private:
    int m_port;
    // The libevent event processor for the primary thread.
    std::unique_ptr<LibeventThread> m_MainBase;
    // The array stores information about each child thread.
    std::vector<std::unique_ptr<LibeventThread>> m_Threads;
    // Custom signal processing.
    std::vector<event*> m_SignalEvents;
};

} // end namespace cs

} // end namespace dzlua

#endif // end DZLUA_CS_LIBEVENT_SERVER_H_
